package com.example.kafkademo.kafkatest1;


import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;

import java.util.*;

public class SparkStreamingWriteRedis {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
//		conf.setMaster("local[4]");
        conf.setMaster("spark://192.168.50070");
        conf.setAppName("transfer App");
        conf.set("spark.streaming.stopGracefullyOnShutdown","true");
        conf.set("spark.default.parallelism", "6");

//        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
        SparkContext create = new SparkContext(conf).getOrCreate();

//        JavaSparkContext javaSparkContext = new JavaSparkContext(spark.sparkContext());
        JavaSparkContext javaSparkContext = JavaSparkContext.fromSparkContext(create);
        JavaStreamingContext jssc = new JavaStreamingContext(javaSparkContext, Durations.seconds(10));

        Map<String, Object> kafkaParams = new HashMap<>();

        kafkaParams.put("bootstrap.servers", "192.168.4.43:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "myGroup998");
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", true);

        Collection<String> topic0 = Arrays.asList("self-topic0");
        Collection<String> topic1 = Arrays.asList("self-topic1");
        Collection<String> topic2 = Arrays.asList("self-topic2");
        Collection<String> topic3 = Arrays.asList("self-topic3");
        Collection<String> topic4 = Arrays.asList("self-topic4");
        Collection<String> topic5 = Arrays.asList("self-topic5");
        List<Collection<String>> topics = Arrays.asList(topic0, topic1, topic2, topic3, topic4, topic5);

        List<JavaDStream<ConsumerRecord<String, String>>> kafkaStreams = new ArrayList<>(topics.size());

//        Collection<String> topic0 = Arrays.asList("self-topic0","self-topic1","self-topic2");

        for (int i = 0; i < topics.size(); i++) {
            kafkaStreams.add(KafkaUtils.createDirectStream(
                    jssc,
                    LocationStrategies.PreferConsistent(),
                    ConsumerStrategies.<String, String>Subscribe(topics.get(i), kafkaParams)));
        }
        JavaDStream<ConsumerRecord<String, String>> stream = jssc.union(kafkaStreams.get(0), kafkaStreams.subList(1, kafkaStreams.size()));

//        stream.foreachRDD((rdd)->{rdd.foreachPartition((crs)->patchTransfer(crs));});
//https://blog.csdn.net/redstarofsleep/article/details/81110319?utm_medium=distribute.pc_relevant_t0.none-task-blog-2%7Edefault%7EBlogCommendFromMachineLearnPai2%7Edefault-1.control&depth_1-utm_source=distribute.pc_relevant_t0.none-task-blog-2%7Edefault%7EBlogCommendFromMachineLearnPai2%7Edefault-1.control
        jssc.start();
        try {
            jssc.awaitTermination();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}
